package cn.tang.tframe.mq.kafka.consumer;

import cn.tang.tframe.common.base.MainConfigLoader;
import cn.tang.tframe.common.utils.io.PropertiesUtil;
import cn.tang.tframe.common.utils.lang.SpiUtils;
import cn.tang.tframe.common.utils.lang.StringUtils;
import cn.tang.tframe.dynconfig.DynConfigRegister;
import cn.tang.tframe.dynconfig.IDynConfigProcessor;
import cn.tang.tframe.mq.IQtMessager;
import cn.tang.tframe.mq.consumer.IQtMsgProcessor;
import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Properties;

/**
 * Created by tanghc on 17/1/20.
 */
public class KafkaConsumerClient {
	static Logger logger = LoggerFactory.getLogger(KafkaConsumerClient.class);
	private KafkaConsumerClient(){}
	private static class KafkaProducerClientHolder{
		private static DelegateKafkaConsumer instance;
		private static Properties properties;
		private static final String producer_config_group = "common";
		private static final String producer_config_dataId = "kafkaConsumer";
		private static final String productCode =
				MainConfigLoader.getInstance().getProductCode();
		private static final String appCode =
				MainConfigLoader.getInstance().getAppCode();
		static {
			DynConfigRegister.getInstance().buildOnce(productCode, appCode, producer_config_group, producer_config_dataId, new IDynConfigProcessor() {
				@Override
				public void procesConfig(String config) {
					properties = PropertiesUtil.loadFromString(config);
				}
			});
		}

		/**
		 * 属性基于0.10.1
		 */
		private static DelegateKafkaConsumer buildKafkaConsumer(String rawTopic,IQtMsgProcessor msgProcessor,Properties prop){
			if(null == prop){
				prop = new Properties();
			}
			Properties allProperties = new Properties();
			allProperties.putAll(properties);
			allProperties.putAll(prop);
//host/port对的列表，用来建立与kafka的初始链接。
//			setDefault(allProperties, "bootstrap.servers", "");//default:
//consumer offsets可以自动提交到kafka的频率（微秒）
//			setDefault(allProperties,"auto.commit.interval.ms","");//default:5000
//当kafka没有初始offset或者server中也不存在任何初始化offset时，consumer遇到这种情况应该从哪里开始获取消息
//			setDefault(allProperties,"auto.offset.reset","");//default:latest
//自动检查消费的消息的CRC32.这个检查保证了消息在发送过程中没有损坏，或者在磁盘上没有损坏。
// 这个检查有可能增加负担，因此对性能要求比较高的情况可能禁用这个检查。
//			setDefault(allProperties,"check.crcs","");//default:true
//
			setDefault(allProperties,"client.id",productCode+"_"+appCode);//default:
//空闲链接的超时时间：server socket处理线程会关闭超时的链接。
//			setDefault(allProperties,"connections.max.idle.ms","");//default:540000
//如果设置为true，则consumer的offset会在后台周期性的上传
//			setDefault(allProperties,"enable.auto.commit","");//default:true
//内部topics（例如offsets）是否需要暴漏给consumer。如果设置true，从内部topic获取数据的方式只能是订阅它。
//			setDefault(allProperties,"exclude.internal.topics","");//default:true
//server针对抓取请求的应答中所包含的最大字节数。
//			setDefault(allProperties,"fetch.max.bytes","");//default:52428800
//server在应答抓取请求之前可以阻塞的最长时间
//			setDefault(allProperties,"fetch.max.wait.ms","");//default:500
//server返回给抓取请求的最小数据量。
//			setDefault(allProperties,"fetch.min.bytes","");//default:1
//代表consumer组的唯一字符串
			setDefault(allProperties,"group.id",StringUtils.join(new Object[]{productCode,appCode},"_"));//default:""
//当使用Kafka的group管理用法时，consumer协作器两次心跳之间的时间间隔。
//			setDefault(allProperties,"heartbeat.interval.ms","");//default:3000
//server返回消息中针对每个partition数据请求的最大数据量。
//			setDefault(allProperties,"max.partition.fetch.bytes","");//default:1048576
//当使用consumer 组管理时，在两次调用poll（）之间的停留时间。
//			setDefault(allProperties,"max.poll.interval.ms","");//default:300000
//一次单独调用poll（）可以返回的消息的最大条数。
//			setDefault(allProperties,"max.poll.records","");//default:500
//更新metadata的时间间隔，
//			setDefault(allProperties,"metadata.max.age.ms","");//default:300000
//用于实现指标统计的类的列表
//			setDefault(allProperties,"metric.reporters","");//default:
//维护计算指标的样本数
//			setDefault(allProperties,"metrics.num.samples","");//default:2
//度量样本的计算的时长
//			setDefault(allProperties,"metrics.sample.window.ms","");//default:30000
//partitions的分配策略的类名。当使用group管理策略时，客户端用来将来将partitions分配给组中consumer实例。
//			setDefault(allProperties,"partition.assignment.strategy","");//default:org.apache.kafka.clients.consumer.RangeAssignor
//TCP接受缓存的大小（SO_RCVBUF）。如果设置为－1，则使用OS默认值.
//			setDefault(allProperties,"receive.buffer.bytes","");//default:65536
//重连给定host之前的等待时间。避免频繁的重连某个host。
//			setDefault(allProperties,"reconnect.backoff.ms","");//default:50
//客户端等待broker应答的超时时间。
//			setDefault(allProperties,"request.timeout.ms","");//default:305000
//重新发送失败请求给某个topic partition之前的最长等待时间，避免极短时间内频繁的重试。
//			setDefault(allProperties,"retry.backoff.ms","");//default:100
//TCP发送的socket的SO_SNDBUF缓存。如果设置为－1，将使用OS的默认值
//			setDefault(allProperties,"send.buffer.bytes","");//default:131072
//当使用Kafka group管理用法时，这个超时时间用来检测consumer是否失效
//			setDefault(allProperties,"session.timeout.ms","");//default:10000
			/**
			 * 以下参数不允许配置
			 */
//			Serializer接口的密钥的类的key
			overrideValue(allProperties, "key.deserializer", "cn.tang.tframe.mq.kafka.serial.QtMessageDeserializer");//defalult:
//			Serializer接口的类的value
			overrideValue(allProperties, "value.deserializer", "cn.tang.tframe.mq.kafka.serial.QtMessageDeserializer");//defalult:
			List<ConsumerInterceptor> consumerInterceptors = SpiUtils.listSpi(ConsumerInterceptor.class);
			StringBuffer intecptStr = new StringBuffer();
			if(null != consumerInterceptors){
				for(int i = 0;i<consumerInterceptors.size();i++){
					if(i != 0){
						intecptStr.append(",");
					}
					intecptStr.append(consumerInterceptors.get(i).getClass().getName());
				}
			}
//			用作拦截器的类的列表。
			overrideValue(allProperties, "interceptor.classes",intecptStr.toString());//defalult:
			return new DelegateKafkaConsumer(rawTopic,new KafkaConsumer<String, IQtMessager>(allProperties),msgProcessor);
		}
		private static void setDefault(Properties properties,String key,Object defalt){
			if(StringUtils.isNotBlank(key) && null != properties && null != defalt){
				Object o = properties.get(key);
				if(null == o){
					o = defalt;
				}
				properties.put(key, o);
			}
		}
		private static void overrideValue(Properties properties,String key,Object newValue){
			if(StringUtils.isNotBlank(key) && null != properties && null != newValue){
				properties.put(key, newValue);
			}
		}
	}

	public static Runnable createConsumerTask(String rawTopic,IQtMsgProcessor msgProcessor,Properties prop){
		DelegateKafkaConsumer delegateKafkaConsumer = KafkaProducerClientHolder.buildKafkaConsumer(rawTopic, msgProcessor, prop);
		if(null == delegateKafkaConsumer){
			throw new RuntimeException("build kafka producer client exception");
		}
		return delegateKafkaConsumer;
	}

}
